package com.at.env2;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 环境
 *
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/9/23 20:20
 */
public class EnvDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8082");
        StreamExecutionEnvironment env = StreamExecutionEnvironment
//                .getExecutionEnvironment(); // 自动识别 是 远程集群，还是idea本地环境
                .getExecutionEnvironment(conf); // conf对象可以去修改一些参数
//                .createLocalEnvironment()
//                .createRemoteEnvironment("master", 8081, "/xxx.jar")

        // 流、批 一体：代码api是同一套，可以指定为 批，也可以指定为 流
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        // 默认是 STREAMING
        // 一般不在代码写死，提交时 参数指定：-Dexecution.runtime.mode=BATCH
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        // TODO 2.读取数据：socket
        env
                .socketTextStream("localhost", 7777)
//                .readTextFile("input/word.txt")
                .flatMap(
                        (String value, Collector<Tuple2<String, Integer>> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(Tuple2.of(word, 1));
                            }
                        }
                )
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(value -> value.f0)
                .sum(1)
                .print();

        // TODO 关于execute总结（了解）
        // 1、默认 env.execute() 触发一个flink job：
        // 一个main方法可以调用多个execute，但是没意义，指定到第一个就会阻塞住
        // 2、env.executeAsync()，异步触发，不阻塞
        //    => 一个main方法里 executeAsync()个数 = 生成的flink job数
        // 3、思考：
        //   yarn-application 集群，提交一次，集群里会有几个flink job：
        //     => 取决于 调用了几个 executeAsync()
        //     => 对应 JobManager 当中，会有n个 JobMaster
//        env.executeAsync();
        env.execute();
    }
}
